package run.liuli.tran.activity.listener;

import lombok.extern.java.Log;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import run.liuli.tran.activity.pojo.Coupon;
import run.liuli.tran.activity.service.CouponService;

import javax.annotation.Resource;
import java.util.List;

@Component
@Log
public class CouponListener {

    @Resource
    private CouponService couponService;


    @KafkaListener(containerFactory = "manualImmediateListenerContainerFactory", topics = "coupon")
    public void grantCoupon(List<Object> message, Acknowledgment ack) {
        message.forEach(meg -> {
            try {
                String[] split = message.toString().replaceAll("[\\[ | \\]]", "").split(",");//[1,2,3]
                Long uid = Long.valueOf(split[0]);
                Long money = Long.valueOf(split[1]);
                Long tid = Long.valueOf(split[2]);
                try {
                    couponService.save(new Coupon(money, uid, tid));
                    ack.acknowledge();
                } catch (DuplicateKeyException e) {
                    log.info("重复消费，tid=" + tid);
                } finally {
                    ack.acknowledge();
                }
            } catch (Exception e) {
                log.warning("消费出错，不返回ack");
            }
        });

    }
}
